package com.fangsheng.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @创建人: 放生
 * @创建时间: 2022/1/22
 * @描述:  rocketmq  事务消息
 */
public class RocketTransaction {


    //---------------MQ的事务 什么是事务----------------------


    /**
     * 还是producer这一侧的
     * 事务消息 消费者端正常案列
     */
    @Test
    public void consumerTransaction() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction_consumer2");
        consumer.setNamesrvAddr("192.168.150.11:9876;192.168.150.12:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("topic_transaction2","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                msgs.forEach(msg-> System.out.println(msg));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();

    }

    @Test
    public void producerTransaction() throws Exception {

        TransactionMQProducer producer = new TransactionMQProducer("transaction_producer2");

        producer.setNamesrvAddr("192.168.10.11:9876;192.168.10.21:9876");

        /**
         * //1,half半消息成功了才能执行本地事务，也需要监听，这里设置了一个事务消息的监听器,会有两个
         * 回调的一个方法，executeLocalTransaction 改方法是在消息发送了一个半消息到broker成功后执行的方法
         * 我们需要把业务逻辑写在这里，比如远程接口调用，和本地mysql的方法执行，执行完以后然后根据执行的结果
         * 返回对应的状态的值，UNKNOW （如果返回的是UNKNOW则会过一会触发rocketmq的回查方法checkLocalTransaction），
         * ROLLBACK_MESSAGE（取消mq中的数据 回滚操作，就不会在触发回查方法） COMMIT_MESSAGE（提交操作，消费者可以消费数据了就不会在触发回查方法）
         *
         * //2，半消息的回调要被监听到
         */
        producer.setTransactionListener(new TransactionListener() {
            //本地事务应该干啥事呀？
            //成本的角度来思考
            //三个地方可以传导业务需要的参数，三种方式传递参数进消息做业务识别的方法
            //1，message 的 body  1，网络带宽，2，污染了body的编码
            //2，userProperty，1，通过网络传递给consumer
            //3，arge方式，local
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

                // 当前确认的方法是我们要写业务逻辑的方法，比如调用另外的一个服务的接口，和 本地数据库数据持久化

                //send ok  half半消息
                    //以下是取出消息 对应上面的放入消息的三个方法
//                String action1 = new String(msg.getBody());
//                String action = msg.getProperty("action");
                String action = (String)arg;
                String transactionId = msg.getTransactionId();
                // transactionId 事务ID 是发送消息事务mq维护的一个事务ID，实际中这个事务的ID要作为本地事务表的一个属性入库
                // 的，然后还要加上一个状态的属性了，这样我们在回查的时候就可以根据查询事务的ID然后根据状态就知道这个事务有没有提
                // 交，这也是回查判断消息是否提交的依据，另外注意这个transactionId要持久化库，不能作为一个标量在当前的方法中，
                // 方法中的局部变量每次方法调用会被刷新调的
                System.out.println("transactionID:" + transactionId);
                /**
                 * 状态有2个：
                 * rocketmq的half半消息，这个状态驱动rocketmq 回查producer
                 * service应该是无状态的，那么应该把transactionId  随着本地事务的执行写入事件状态表
                 */
                switch (action){
                    case "0":
                        System.out.println(Thread.currentThread().getName()+"send half：Async api call...action ： 0");
//                        HelloSys.call(ooxx,sdfsdf);
                        return LocalTransactionState.UNKNOW;  //rocketmq 会回调我去check
                    case "1":
                        System.out.println(Thread.currentThread().getName()+"send half：localTransaction faild...action ： 1");
                        /*
                        transaction.begin
                        throw...
                        transaction.rollback
                         */
                        //观察 consumer是消费不到 rollback的message的
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    case "2":
                        System.out.println(Thread.currentThread().getName()+"send half：localTransaction ok...action ： 2");
                        /*
                        transaction.begin
                        throw...
                        transaction.commit..ok
                         */
                        //观察 consumer是肯定消费的到的，只不过，还要验证，中途会不会做check
                        try {
                            Thread.sleep(1000*10);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return LocalTransactionState.COMMIT_MESSAGE;
                }


                return null;
            }

            // 当前方法是mq 回查调用的方法，回查的时候我们可以查询本地事务表的id 然后根据状态来区分当前的事务有没有提
            // 交，如果没有我们就继续返回LocalTransactionState.UNKNOW;那么下次（过一会）还会走回查，默认同一个事务iD
            // 在一直是返回UNKNOW的情况下，默认会一直回调18次，后面就不会在调用，如果觉得太多也可以自行设置次数，如果调用了
            // 2次后还没到总次数，当前服务挂了，下次服务启动只要还是UNKNOW，那么下次服务重启也会继续调用，因为这些调用的次数，
            // 状态等元数据rocketmq是维护在 rocketmq 的broker端的。
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                //call back check

                String transactionId = msg.getTransactionId();
                String action = msg.getProperty("action");
                int times = msg.getReconsumeTimes();


                switch (action){
                    case "0" :
                        System.out.println(Thread.currentThread().getName()+"check：action ：0,UNKNOW: "+ times);


                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return LocalTransactionState.COMMIT_MESSAGE;


                    case "1" :
                        System.out.println(Thread.currentThread().getName()+"check：action ：1 ROLLBACK"+ times);
                        //按理来说就是观察事务表
                        return LocalTransactionState.UNKNOW;
                    case "2" :
                        System.out.println(Thread.currentThread().getName()+"check：action ：2 COMMIT"+ times);

                        //check 都是观察事务表的
                        return LocalTransactionState.UNKNOW;
                    //未来你盯着事务表得到结果  COMMIT  ROLLBACK

                    //做了一个长时间的事务：check是要做的
                    //如果produer重启了一下，还能不能响应这个检查

                }


                return null;
            }
        });

        //need thread  ,不是必须要去配置的
        producer.setExecutorService(new ThreadPoolExecutor(
                1,
                Runtime.getRuntime().availableProcessors(),
                2000,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {  //loop worker  -> 消费你提供的queue中的runnable（task）
                        return new Thread(r,"Transaction thread ");
                    }
                }
        ));

        producer.start();

//        for (int i = 0; i < 10; i++) {
//
//            Message msgt =  new Message(
//                    "topic_transaction2",
//                    "TagT",
//                    "key:"+i,
//                    ("message : " + i).getBytes()
//            );
//            msgt.putUserProperty("action",i%3+"");
//
//
//            //发送的是半消息
//            TransactionSendResult res = producer.sendMessageInTransaction(msgt,i%3+"");
//            System.out.println(res);
//        }
        System.in.read();

    }










}
